package com.willie.elasticjob.runner;

import java.util.List;

import javax.annotation.Resource;

import com.dangdang.ddframe.job.api.ElasticJob;
import com.dangdang.ddframe.job.api.dataflow.DataflowJob;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.JobTypeConfiguration;
import com.dangdang.ddframe.job.config.dataflow.DataflowJobConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import com.dangdang.ddframe.job.lite.spring.api.SpringJobScheduler;
import com.willie.elasticjob.dao.ElasticJobConfigDao;
import com.willie.elasticjob.entity.ElasticJobConfig;
import com.willie.elasticjob.factory.AbstractElasticJobFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import com.dangdang.ddframe.job.event.JobEventConfiguration;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
import org.springframework.util.CollectionUtils;

/**
 * 定时任务配置
 *
 * @author xing
 */
@Component
@Slf4j
public class ElasticJobRunner implements CommandLineRunner {
    @Resource
    private ZookeeperRegistryCenter registryCenter;

    @Resource
    private JobEventConfiguration jobEventConfiguration;

    @Resource
    private ElasticJobConfigDao elasticJobConfigDao;

    @Override
    public void run(String... args) {
        List<ElasticJobConfig> elasticJobConfigList = elasticJobConfigDao.findAll();
        if (CollectionUtils.isEmpty(elasticJobConfigList)) {
            return;
        }
        elasticJobConfigList.forEach(this::startSchedule);
    }

    /**
     * 启动定时任务
     *
     * @param jobConfig 定时任务配置
     */
    private void startSchedule(ElasticJobConfig jobConfig) {
        try {
            //创建任务
            ElasticJob job = AbstractElasticJobFactory.createElasticJob(jobConfig.getJobClass());
            //配置任务
            LiteJobConfiguration config = LiteJobConfiguration.newBuilder(jobConfiguration(job, jobConfig)).overwrite(true).build();
            SpringJobScheduler jobScheduler = new SpringJobScheduler(job, registryCenter, config, jobEventConfiguration);
            jobScheduler.init();
            log.info("初始化定时任务 {} ", jobConfig.toString());
        } catch (Exception e) {
            log.error("注册Job出错：{} ", jobConfig.toString(), e);
        }
    }

    /**
     * 定时任务配置
     *
     * @param job       任务
     * @param jobConfig 任务配置
     * @return
     */
    private JobTypeConfiguration jobConfiguration(ElasticJob job, ElasticJobConfig jobConfig) {
        JobCoreConfiguration jobCoreConfiguration = JobCoreConfiguration
                .newBuilder(jobConfig.getJobName(), jobConfig.getCron(),
                        jobConfig.getShardingTotalCount())
                .shardingItemParameters(jobConfig.getShardingItemParameters())
                .misfire(Boolean.valueOf(jobConfig.getMisfire()))
                .description(jobConfig.getDescription())
                .failover(Boolean.valueOf(jobConfig.getFailover()))
                .jobParameter(jobConfig.getJobParameter())
                .build();
        if (job instanceof SimpleJob) {
            return new SimpleJobConfiguration(jobCoreConfiguration, job.getClass().getCanonicalName());
        }

        if (job instanceof DataflowJob) {
            return new DataflowJobConfiguration(jobCoreConfiguration, job.getClass().getCanonicalName(),
                    Boolean.valueOf(jobConfig.getStreamingProcess()));
        }
        throw new RuntimeException("未知类型定时任务：" + job.getClass().getName());
    }
}
